package com.sunday.common.reactor.e1_Subscribe;

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.Disposables;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * https://projectreactor.io/docs/core/release/reference/#core-features
 * <p>
 * 这是一些 Reactor 库中用于创建 Flux 的方法的比较：
 * <p>
 * Flux.push 在 Reactor 3.4 版本中被引入，它允许您根据需要以异步或同步方式推送元素给下游。
 * 每次推送都需要指定一个 Signal（信号），可以是 onNext、onError 或 onComplete。
 * <p>
 * Flux.generate 需要提供一个初始状态和一个生成器函数，该函数在每次请求数据时会被调用。
 * 生成器函数可以发出一个元素、完成流或者发出错误信号。适合有状态的生成逻辑。
 * <p>
 * Flux.create 接受一个消费者，该消费者可以向下游发送任意数量的元素、完成信号或错误信号。
 * 比较自由地生成事件流，通常使用了回压支持，并且可以采用不同的生成逻辑。
 * <p>
 * Flux.just 创建一个包含指定元素的 Flux，最多可包含 9 个元素。
 * 通常用于从已知的元素集合中创建 Flux（静态创建）。
 * 总体而言，这些方法在如何生成和推送元素以及对状态的处理上有所不同。Flux.push 是推送元素的方法，Flux.generate 和 Flux.create 则更专注于生成元素的方式，而 Flux.just 则用于静态创建包含指定元素的 Flux。
 */
@Slf4j
public class CreateFluxOrMonoAndSubscribe {

    /**
     * 要创建 的序列，您可以枚举它们或将它们放入 并从中创建 Flux
     */
    @Test
    public void create() {

        Flux<String> seq1 = Flux.just("foo", "bar", "foobar");
        log.info("{}", seq1); // 输出：reactor.core.publisher.FluxArray

        List<String> iterable = Arrays.asList("foo", "bar", "foobar");
        Flux<String> seq2 = Flux.fromIterable(iterable);
        log.info("{}", seq2); // 输出：reactor.core.publisher.FluxIterable

        Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3);
        log.info("{}", numbersFromFiveToSeven); // 输出：reactor.core.publisher.FluxRange

        Mono<String> noData = Mono.empty();
        log.info("{}", noData); // 输出：reactor.core.publisher.MonoEmpty

        Mono<String> data = Mono.just("foo");
        log.info("{}", data); // 输出：reactor.core.publisher.MonoJust

    }

    /**
     * 订阅
     * 在订阅方面，请使用 Java 8 lambda。你 有多种变体可供选择，这些变体将 lambda 用于不同的 回调的组合，
     * 如以下方法签名所示：FluxMono.subscribe()
     * <p>
     * subscribe();
     * 订阅并触发序列。
     * <p>
     * subscribe(Consumer<? super T> consumer);
     * 对每个产生的值做一些事情。
     * <p>
     * subscribe(Consumer<? super T> consumer,
     * Consumer<? super Throwable> errorConsumer);
     * 处理值，但也要对错误做出反应。
     * <p>
     * subscribe(Consumer<? super T> consumer,
     * Consumer<? super Throwable> errorConsumer,
     * Runnable completeConsumer);
     * 处理值和错误，但也会在序列成功时运行一些代码 完成。
     * <p>
     * 这些变体返回对订阅的引用，您可以使用该订阅取消 在不需要更多数据时订阅。
     * 取消后，源应停止 产生价值并清理它创造的任何资源。此取消和清理行为 在 Reactor 中由通用接口表示。Disposable
     */
    @Test
    public void subscribe() {

        // 设置一个在订阅者连接时生成三个值。Flux
        Flux<Integer> ints = Flux.range(1, 3);

        log.info("=====================     subscribe()    ===========================");
        //以最简单的方式订阅。
        ints.subscribe();

        log.info("=====================     subscribe(Consumer<? super T> consumer)    ===========================");
        // 使用将打印值的订阅者进行订阅。
        ints.subscribe(i -> log.info("{}", i));

        log.info("=====================     subscribe(Consumer<? super T> consumer, Consumer<? super Throwable> errorConsumer)    ===========================");
        // 设置一个 Flux，该 Flux 在订阅者连接时生成四个值。
        Flux<Integer> errInts = Flux.range(1, 4)
                // 我们需要一个地图，以便我们可以以不同的方式处理某些值。
                .map(i -> {
                    // 对于大多数值，请返回该值。
                    if (i <= 3) return i;
                    // 但是，对于 4，我们将抛出一个异常。
                    throw new RuntimeException("Got to 4");
                });
        // 使用将打印值的订阅者和将打印错误的错误处理程序进行订阅。
        errInts.subscribe(i -> log.info("{}", i),
                error -> log.error("Error {}", error.toString()));

        log.info("=====================     subscribe(Consumer<? super T> consumer, Consumer<? super Throwable> errorConsumer, Runnable completeConsumer)    ===========================");
        // 处理值和错误，但也会在序列成功时运行一些代码 完成。
        // 错误信号和完成信号都是终端事件，并且不包括一个 另一个（你永远不会两者兼而有之）。要使消费者工作完成，我们必须注意不要 触发错误。
        ints.subscribe(i -> log.info("{}", i),
                error -> log.error("Error {}", error.toString()),
                // 使用包含完成事件处理程序的订阅服务器进行订阅。
                () -> log.info("Done"));
        errInts.subscribe(i -> log.info("[err]{}", i),
                error -> log.error("[err]Error {}", error.toString()),
                // 使用包含完成事件处理程序的订阅服务器进行订阅。
                () -> log.info("[err]Done"));

        log.info("=====================     subscribe(Consumer<? super T> consumer, Consumer<? super Throwable> errorConsumer, Runnable completeConsumer, Consumer<? super Subscription> subscriptionConsumer)    ===========================");

    }

    /**
     * subscribe()Disposable
     */
    @Test
    public void subscribeDisposable() {
        Flux<Integer> ints = Flux.range(1, 5);
        // 一个新的{@link Disposable}，可用于取消底层的{@link Subscription}
        Disposable disposable = ints.subscribe(i -> log.info("{}", i));
        /**
         * 在 Reactor 中，当你使用 subscribe 方法来订阅一个 Flux 或 Mono 时，会返回一个 Disposable 对象，该对象代表了订阅，而不是资源本身。
         * 在典型情况下，一旦订阅完成，该 Disposable 会被自动处置。
         * 因此，即使你没有手动调用 dispose 方法，只要订阅完成，disposable.isDisposed() 往往会返回 true。
         * 这种行为是 Reactor 设计的一部分，它简化了资源管理，确保释放资源的可靠性。
         * 如果你需要手动处置 Disposable，你可以调用 dispose 方法。但很多情况下，这并非必须，因为 Reactor 会负责在适当的时机进行处置。
         */
        log.info("[disposable.isDisposed] 之前 ：{}", disposable.isDisposed()); // 输出：true
        disposable.dispose();
        log.info("[disposable.isDisposed] 之后 ：{}", disposable.isDisposed()); // 输出：true

        // composite 可以将多个 Disposable 合并为一个
        Disposable.Composite composite = Disposables.composite(disposable);
        log.info("[composite.isDisposed] 之前 ：{}", composite.isDisposed()); // 输出：false
        composite.dispose();
        log.info("[composite.isDisposed] 之后 ：{}", composite.isDisposed()); // 输出：true

        Disposable disposable1 = ints.subscribe(i -> log.info("{}", i));

        /**
         * Disposables.swap()函数的作用是将当前对象所持有的资源引用与另一个对象所持有的资源引用进行交换。
         * 这意味着当前对象将释放其所持有的资源引用，并将该引用赋给另一个对象，同时另一个对象将其资源引用赋给当前对象。
         */
        Disposable.Swap swap = Disposables.swap();
        /**
         * 交换后，disposable1 释放了资源，而 swap 持有了资源
         */
        swap.replace(disposable1);
        log.info("[swap.isDisposed] 之前 ：{}", swap.isDisposed()); // 输出：false
        swap.dispose();
        log.info("[swap.isDisposed] 之后 ：{}", swap.isDisposed()); // 输出：true

    }

    /**
     * 4.3.3. Lambda 的替代方案：BaseSubscriber
     */
    @Test
    public void baseSubscriber() {
        BaseSubscriber<Integer> ss = new BaseSubscriber<>() {

            /**
             * 在订阅时调用此方法，它将触发对给定订阅的请求。
             * @param subscription the subscription to optionally process
             */
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                log.info("Subscribed ：{}", subscription);
                /**
                 * 在给出的接口 Subscription 中，cancel() 和 request(long n) 方法分别用于处理订阅的取消和请求。具体作用如下：
                 *
                 * request(long n) 方法：用于向发布者（Publisher）请求数据。
                 * 通过这个方法，订阅者可以要求从发布者那里获取指定数量的元素。参数 n 是一个严格正数，表示要请求的元素数量。
                 * 订阅者可以通过多次调用此方法来反复请求元素。
                 *
                 * cancel() 方法：用于取消订阅。
                 * 当订阅者调用此方法时，它要求发布者停止发送数据并清理资源。需要注意的是，
                 * 即使调用了 cancel() 方法，之前已经请求的数据可能仍然会被发送，以满足先前已经发出的需求。
                 */
                request(1);
            }

            /**
             * 当接收到一个订阅的数据时调用此方法。
             * @param value the emitted value to process
             */
            @Override
            protected void hookOnNext(Integer value) {
                log.info("value ：{}", value);
                request(1);
            }

            /**
             * 当接收到一个订阅的错误时调用此方法。
             * @param throwable the error to process
             */
            @Override
            protected void hookOnError(Throwable throwable) {
                log.error("Error ：{}", throwable.toString());
            }

            /**
             * 当接收到一个订阅的完成信号时调用此方法。
             */
            @Override
            protected void hookOnCancel() {
                log.info("Cancel");
            }

            /**
             * 当接收到一个订阅的完成信号时调用此方法。
             */
            @Override
            protected void hookOnComplete() {
                log.info("Complete");
            }
        };

        Flux<Integer> ints = Flux.range(1, 4);
        ints.subscribe(ss);
        ss.cancel();
    }

    /**
     * 4.3.4. 关于背压和重塑请求的方法
     * <p>
     * 在 Reactor 中实现背压时，将用户压力传播回源的方式是将 a 发送到上游运营商。 当前请求的总和有时称为当前“需求”或“待处理请求”。
     * 需求上限为 ，表示无限制的请求（意思是“尽可能快地生产”——基本上禁用背压）。requestLong.MAX_VALUE
     * 第一个请求来自订阅时的最终订阅者，但最直接的订阅方式会立即触发以下无限制请求：Long.MAX_VALUE
     * subscribe()以及它的大多数基于 lambda 的变体（除了具有Consumer<Subscription>)
     * block()和blockFirst()blockLast()
     * 遍历 ORtoIterable()toStream()
     */
    @Test
    public void backpressure() {
        Flux<Integer> flux = Flux.range(1, 10);

        flux
                // doOnRequest 方法用于在每次请求时执行一些操作。
                .doOnRequest(r -> log.info("request of " + r))
                .subscribe(new BaseSubscriber<>() {
                    @Override
                    public void hookOnSubscribe(Subscription subscription) {
                        request(1);
                    }

                    @Override
                    public void hookOnNext(Integer integer) {
                        log.info("Cancelling after having received " + integer);
//                        request(1);
                        cancel();
                    }
                });

        // blockFirst()方法是一个阻塞调用，它将等待序列中的第一个元素并返回它。
        log.info("blockFirst {}", flux.blockFirst());
        // blockLast()方法是一个阻塞调用，它将等待序列中的最后一个元素并返回它。
        log.info("blockLast {}", flux.blockLast());

    }

    /**
     * 4.4.1. generate
     * 不允许多次调用 sink.next(T)方法，因为它会导致 IllegalStateException。
     */
    @Test
    public void synchronousGenerate() {

        log.info("generate(Consumer<SynchronousSink<T>> generator) ================================================");

        Flux.generate(sink -> {
            sink.next("Hello");
            // java.lang.IllegalStateException: More than one call to onNext
//            sink.next("Hello");
            sink.complete();
        }).subscribe(i -> log.info("{}", i));

//        log.info("fromStream(Supplier<Stream<? extends T>> streamSupplier) ================================================");
//        Flux.fromStream(() -> Stream.of("foo", "bar"))
//                .subscribe(i -> log.info("{}", i));

        log.info("基于状态的示例 generate(Supplier<S>, BiFunction)================================================");
        Flux<String> flux = Flux.generate(
                // 我们提供初始状态值 0。
                () -> 0,
                // 我们提供一个生成器函数，它将状态值作为输入，并返回下一个值。
                (state, sink) -> {
                    // 我们使用 sink.next(T)方法来生成下一个值。
                    sink.next("3 x " + state + " = " + 3 * state);
                    // 我们还使用它来选择何时停止。
                    if (state == 10) sink.complete();
                    // 我们返回一个在下一次调用中使用的新状态（除非 序列在此终止）
                    return state + 1;
                });
        flux.subscribe(i -> log.info("{}", i));

        log.info("可变状态变体 generate(Supplier<S>, BiFunction)================================================");
        flux = Flux.generate(
                // 这一次，我们生成一个可变对象作为状态。
                AtomicLong::new,
                (state, sink) -> {
                    // 我们在这里改变状态。
                    long i = state.getAndIncrement();
                    sink.next("3 x " + i + " = " + 3 * i);
                    if (i == 10) sink.complete();
                    // 我们返回与新状态相同的实例。
                    return state;
                });
        flux.subscribe(i -> log.info("{}", i));

        log.info("generate(Supplier<S>, BiFunction, Consumer<S>) ===============================================");
        flux = Flux.generate(
                // 我们提供初始状态值 0。
                () -> 0,
                // 我们提供一个生成器函数，它将状态值作为输入，并返回下一个值。
                (state, sink) -> {
                    // 我们使用 sink.next(T)方法来生成下一个值。
                    sink.next("3 x " + state + " = " + 3 * state);
                    // 我们还使用它来选择何时停止。
                    if (state == 10) sink.complete();
                    // 我们返回一个在下一次调用中使用的新状态（除非 序列在此终止）
                    return state + 1;
                },
                // 我们还可以提供一个消费者，它将在序列终止时接收最终状态。
                (state) -> log.info("state: {}", state));
        flux.subscribe(i -> log.info("{}", i));

    }

    /**
     * 4.4.2. create
     * create是一种更高级的程序化创建形式，它是 适用于每轮多次发射，甚至来自多个线程。Flux
     */
    @Test
    public void asynchronousAndMultithreadedCreate() throws InterruptedException {

        log.info("create(Consumer<FluxSink<T>> emitter)================================================");
        Flux<String> flux = Flux.create(sink -> {
            sink.next("Hello");
            sink.next("World");
            sink.complete();
        });
        flux.subscribe(i -> log.info("{}", i));

        /**
         * IGNORE           完全忽略下游背压请求。 当队列在下游已满时，这可能会产生这种情况。IllegalStateException
         * ERROR            在下游无法保持时发出信号 向上。IllegalStateException
         * DROP             在下游尚未准备好接收时丢弃输入信号 它。
         * LATEST           让下游只从上游获取最新信号。
         * BUFFER（默认值）   在下游无法跟上时缓冲所有信号。 （这会导致无限制的缓冲，并可能导致 ）。OutOfMemoryError
         */
        log.info("create(Consumer<FluxSink<T>> emitter, FluxSink.OverflowStrategy backpressure)================================================");
        flux = Flux.create(sink -> {
            for (int i = 0; i < 5; i++) {
                sink.next("Hello");
            }
            sink.complete();
        }, FluxSink.OverflowStrategy.BUFFER);
        flux.subscribe(i -> log.info("{}", i));

        log.info("create(Consumer<? super FluxSink<T>> emitter)================================================");
        flux = Flux.create(sink -> {
            log.info("create");
            // onRequest()方法用于处理下游请求。
            sink.onRequest(request -> {
                log.info("request: {}", request);
                sink.next("Hello");
            });
            // onCancel()方法用于处理下游取消。
            sink.onDispose(() -> {
                log.info("dispose");
            });
        });
        flux.subscribe(i -> log.info("{}", i));
    }

    /**
     * 混合推/拉模型
     */
    @Test
    public void pushAndPull() {
        log.info("thread================================================");
        CountDownLatch latch = new CountDownLatch(5);
        Flux<String> flux = Flux.create(sink -> {
            AtomicLong atomicLong = new AtomicLong();
            // 启动五条线程，每条线程都会生成五个值，并且休眠1s。
            for (int i = 0; i < 5; i++) {
                new Thread(() -> {
                    try {
                        long pointer = atomicLong.getAndIncrement();
                        log.info("thread : {}, pointer : {}", Thread.currentThread().getName(), pointer);
                        if (pointer == 3) {
                            log.info("[sleep] 5s");
                            TimeUnit.SECONDS.sleep(5);
                            log.info("[open] 5s");
                        }
                        for (int j = 0; j < 5; j++) {
                            sink.next(Thread.currentThread().getName() + "->" + pointer + "->" + j);
                            TimeUnit.SECONDS.sleep(1);
                            /**
                             * 重点 1
                             * 如果我们在这里抛出一个异常，它将被传播到订阅者。
                             * 被异常输出后，订阅者将不再接收任何值。
                             */
//                            if(j == 2){
//                                sink.error(new RuntimeException("error"));
//                            }
                        }
                        latch.countDown();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }).start();
            }
            /**
             * 混合推/拉模型
             * 大多数 Reactor 算子，比如 ，都遵循混合推/拉模型。 我们的意思是，尽管大多数处理都是异步的 （建议一种推送方法），它有一个小的拉动组件： 请求。create
             * 使用者从源中提取数据，因为它不会发出任何内容 直到第一次请求。源将数据推送给消费者，只要它 变得可用，但在其请求量的范围内。
             * 请注意，两者都允许设置使用者 为了管理请求量并确保数据被推送 仅当有待处理的请求时才使用接收器。
             */
            sink.onRequest(request -> {
                log.info("[onRequest]request: {}", request);
                for (int j = 0; j < 10; j++) {
                    sink.next("Hello" + j);
                }
            }).onCancel(() -> {
                log.info("[onCancel]dispose");
            }).onDispose(() -> {
                log.info("[onDispose]dispose");
            });

            /**
             * 重点 3
             * 等待所有线程执行完毕
             * 如果我们不等待，那么订阅者将不会接收到所有值。
             * 因为我们在订阅时，线程还没有执行完毕。
             * 所以我们需要等待所有线程执行完毕，再通知订阅者。
             */
            try {
                latch.await();
            } catch (InterruptedException e) {
                sink.error(new RuntimeException(e));
            }
            sink.complete();

        });
        log.info("{}", flux);
        Disposable disposable = flux.subscribe(i -> log.info("[subscribe]{}", i), throwable -> log.error(throwable.toString()));
        log.info("{}", disposable.isDisposed());
//        /**
//         * 重点 2
//         * 手动取消订阅
//         */
//        // 取消订阅
//        disposable.dispose();
//        // 由于我们在订阅时取消了订阅，因此我们不会看到所有值。
//        log.info("{}", disposable.isDisposed());
//        TimeUnit.SECONDS.sleep(10);
    }

    /**
     * 4.4.3. push
     * push是介于 和 之间的中间地带，适合 处理来自单个生产者的事件。
     * 从某种意义上说，它类似于 它也可以是异步的，并且可以使用任何 支持的溢出策略。但是，一次只能调用一个生成线程。
     */
    @Test
    public void asynchronousButSingleThreadedPush() {

        log.info("push(Consumer<FluxSink<T>> generator)================================================");
        Flux<String> flux = Flux.push(sink -> {
            sink.next("Hello");
            sink.next("World");
            sink.complete();
        });
        flux.subscribe(i -> log.info("{}", i));

        log.info("push(Consumer<FluxSink<T>> generator, FluxSink.OverflowStrategy backpressure)================================================");
        flux = Flux.push(sink -> {
            for (int i = 0; i < 5; i++) {
                sink.next("Hello");
            }
            sink.complete();
        }, FluxSink.OverflowStrategy.BUFFER);
        flux.subscribe(i -> log.info("{}", i));

        log.info("push(Consumer<? super FluxSink<T>> generator)================================================");
        flux = Flux.push(sink -> {
            log.info("create");
            // onRequest()方法用于处理下游请求。
            sink.onRequest(request -> {
                log.info("request: {}", request);
                sink.next("Hello");
            });
            // onCancel()方法用于处理下游取消。
            sink.onDispose(() -> {
                log.info("dispose");
            });
        });
        flux.subscribe(i -> log.info("{}", i));
    }

    /**
     * 在 Reactor 中，Flux.create 和 Flux.push 两者之间的主要区别在于以下几点：
     * <p>
     * 灵活性:
     * <p>
     * Flux.create 提供了一个更为灵活的方式来创建 Flux，你可以在其中使用 FluxSink 手动发射元素，并且可以定义对取消信号的处理。
     * Flux.push 则提供了一种更简化的方式来向数据流中推入元素，适用于那些不需要太多自定义逻辑的情况。
     * <p>
     * 底层控制：
     * <p>
     * 在 Flux.create 中，你可以完全控制元素的产生和结束信号的发送，因为它提供了一个 FluxSink 对象，给予你更多底层的控制能力。
     * Flux.push 则更加简化，允许你向已经提供的 Emitter 对象推送元素，但不提供像 FluxSink 那样的底层控制。
     * 总体来说，Flux.create 更适用于需要更多自定义控制的场景，而 Flux.push 则更便捷，适用于那些不需要太多自定义逻辑的情形。
     */


    /**
     * 4.4.4. 句柄
     * 该方法有点不同：它是一个实例方法，这意味着 它被链接到现有源上（就像常见的运算符一样）。它存在 在 和 .handleMonoFlux
     * 从某种意义上说，它接近于 ，因为它使用了 和 只允许逐个排放。但是，可用于生成 每个源元素的任意值，可能会跳过某些元素。在 这样，它可以作为 和 的组合。
     */
    @Test
    public void handle() {
        Flux<String> flux = Flux.just("foo", "bar", "foobar");
        flux.handle((s, sink) -> {
            if (s.length() == 3) {
                sink.next(s);
            }
        }).subscribe(i -> log.info("{}", i));
    }


}
